package com.hexinfo.dmpro.sparing.service.impl;


import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.file.FileAppender;
import com.hexinfo.dmpro.common.utils.ConnExecuteUtil;
import com.hexinfo.dmpro.sparing.dto.DistCPDTO;
import com.hexinfo.dmpro.sparing.service.DistCPTimeService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Service
@Slf4j
public class DistCPTimeServiceImpl implements DistCPTimeService {

    @Value("${hdfs.heatBackup}")
    private String heatBackup;

    //hdfs热备完成数量
    private Map<String, Integer> statusMap = new ConcurrentHashMap<String, Integer>();
    private synchronized void updateStatus(String tableName) {
        statusMap.put(tableName, statusMap.getOrDefault(tableName,0)+1);
    }

    //表完成数量
    private Map<String, Integer> tableSizeMap = new ConcurrentHashMap<String, Integer>();
    private synchronized void addTableSize(String souName) {
        tableSizeMap.put(souName, tableSizeMap.getOrDefault(souName,0)+1);
    }

    //任务结束标志
    private Map<String, Boolean> endFlagMap = new ConcurrentHashMap<String, Boolean>();
    //修改结束状态
    public synchronized void updateEndFlag(String souName,Boolean state) {
        endFlagMap.put(souName, state);
    }
    //获取结束状态
    public Boolean getEndFlag(String souName) {
        return endFlagMap.get(souName);
    }
    //删除key
    public synchronized void deleteEndFlag(String souName) {
        endFlagMap.remove(souName);
    }

    @Override
    @Async("threadPoolTaskExecutor")
    public void execShell(DistCPDTO distCPDTO,int sizes) {
        //有无源分类(1：有源分区字段，2：无源分区字段（时间窗口）)
        String shellTar = heatBackup + " " + distCPDTO.getBackToTar();
        //第二步：源--》备份--》目标
        int b = ConnExecuteUtil.execShellCode(shellTar);
        if (b == 0){
            //distCPDTO.getTableLog().append(logTime("shell命令："+shellBak+"--执行结果："+a+"--hdfs文件大小："+distCPDTO.getFileSize()));
        } else {
            logLog(distCPDTO.getHeatlog(),distCPDTO.getTableLog(),"无源（时间窗口）执行错误shell：" + shellTar + "  执行结果：" + b,"1");
        }
        timeTableIsFinish(distCPDTO,sizes);
    }

    /**
     *  时间窗口
     */
    private void timeTableIsFinish(DistCPDTO distCPDTO,int sizes){
        // 更新指定表完成情况
        updateStatus(distCPDTO.keyMap());
        // 判断指定表所有任务是否都完成
        if (statusMap.get(distCPDTO.keyMap()).equals(distCPDTO.getHdfsFileSize())){
            statusMap.remove(distCPDTO.keyMap());  //移除key
            addTableSize(distCPDTO.getSourceDataName());    //表完成数量加一
            Integer integer = tableSizeMap.get(distCPDTO.getSourceDataName());
            logLog(distCPDTO.getHeatlog(),distCPDTO.getTableLog(),"【"+distCPDTO.keyMap()+"同步完成】--"+integer,"1");
            distCPDTO.getTableLog().flush();     //刷进日志中
            logLog(distCPDTO.getHeatlog(),distCPDTO.getTableLog(),"","2"); //刷进日志中
            if (integer.equals(sizes) || integer == sizes){
                tableSizeMap.remove(distCPDTO.getSourceDataName());  //移除key
                logLog(distCPDTO.getHeatlog(),distCPDTO.getTableLog(),"【全部表同步完成】","1"); //刷进日志中
                logLog(distCPDTO.getHeatlog(),distCPDTO.getTableLog(),"","2"); //刷进日志中
                updateEndFlag(distCPDTO.getSourceDataName(),true);  //修改任务状态
            }
        }
    }

    /**
     * 加锁，防止并发问题
     * @param heatlog
     * @param tableLog
     * @param logStr
     * @param str
     */
    private synchronized void logLog(FileAppender heatlog, FileAppender tableLog, String logStr ,String str) {
        if (str.equals("1")){
            heatlog.append(logTime(logStr));
            tableLog.append(logTime(logStr));
        }else {
            heatlog.flush();     //刷进日志中
        }
    }

    /**
     * 日志加日期时间
     * @param str
     * @return
     */
    private String logTime(String str) {
        return "["+DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"]" +str;
    }

}
